package com.example.kafka.service;

import com.example.kafka.config.KafkaConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Service;

import java.util.Properties;

/**
 * 关键业务 acks=0 Service
 */
@Service
public class AcksLeakService {
    public void send1(String msg) {
        Properties props = KafkaConfig.getProducerProps();
        props.put("acks", "0");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        try {
            producer.send(new ProducerRecord<>("topic_acks", msg));
        } finally {
            producer.close();
        }
        // 未关闭 producer
    }

    public void send2(String msg) {
        Properties props = KafkaConfig.getProducerProps();
        props.put("acks", "all");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>("topic_acks", msg));
        // 未关闭 producer
    }

    public void send3(String msg) {
        Properties props = KafkaConfig.getProducerProps();
        props.put("acks", "1");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>("topic_acks", msg));
        // 未关闭 producer
    }

    public void send4(String msg) {
        Properties props = KafkaConfig.getProducerProps();
        props.put("acks", "0");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>("topic_acks", msg));
        // 未关闭 producer
    }

    public void send5(String msg) {
        Properties props = KafkaConfig.getProducerProps();
        props.put("acks", "all");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>("topic_acks", msg));
        // 未关闭 producer
    }
}
